package zhongbo.java.air.listener;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import zhongbo.java.air.monitor.model.AirAlertLog;
import zhongbo.java.air.monitor.service.IAirAlertLogService;
import zhongbo.java.air.monitor.web.websocket.AlertZoneWebSocket;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

@Component
@Slf4j
public class RedisSubscribeLogChannelListener implements MessageListener {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private AlertZoneWebSocket alertZoneWebSocket; // 注入 AlertZoneWebSocket

    @Autowired
    private IAirAlertLogService<AirAlertLog> airAlertLogService; // 注入 IAirAlertLogService

    // 使用 ConcurrentHashMap 来存储频道和处理器的映射关系
    private final Map<String, Consumer<String>> messageHandlers = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // 初始化时可以订阅一些默认频道
        subscribe("flight_activity_channel", message -> {
            log.info("处理飞行活动消息: {}", message);
        });
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = new String(message.getChannel());
        String body = new String(message.getBody());
        log.info("从频道 {} 收到消息: {}", channel, body);
        cn.hutool.json.JSONObject json = JSONUtil.parseObj(body);

        // 获取 icao 并记录日志
        String icao = json.getStr("icao_code");
        log.info("获取到的 icao: {}", icao);

        // 构建 Redis 缓存键
        String redisKey = "alert:history:" + icao;
        log.info("生成的 Redis 键: {}", redisKey);

        // 检查 Redis 缓存
        String recentAlertsJson = stringRedisTemplate.opsForValue().get(redisKey);
        if (recentAlertsJson != null) {
            log.info("找到5分钟内的告警日志记录，结束告警判断");
            return;
        }

        // 调用 AlertZoneWebSocket 的方法
        alertZoneWebSocket.checkAndBroadcastAlert(json);

        // 获取相应的处理器并处理消息
        Consumer<String> messageHandler = messageHandlers.get(channel);
        if (messageHandler != null) {
            messageHandler.accept(body);
        } else {
            log.warn("没有找到频道 {} 的处理器", channel);
        }
    }

    // 添加 subscribe 方法
    public void subscribe(String channel, Consumer<String> messageHandler) {
        log.info("订阅频道: {}", channel);
        messageHandlers.put(channel, messageHandler);
        stringRedisTemplate.getConnectionFactory().getConnection().subscribe(this, channel.getBytes());
    }
}